package com.edu.yx.app.App_03_DwdApp.Dwd_06_Play;

import com.edu.yx.app.APP_01_BaseApp.BaseSqlApp;
import com.edu.yx.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static com.edu.yx.common.Constant.*;

public class Dwd_01_Play_VideoDetail extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_01_Play_VideoDetail().init(
                3061,
                2,
                "Dwd_01_Play_VideoDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        //1.读取视频维度信息
        readVideoInfo(tEnv);
//        tEnv.sqlQuery("select * from video_info").execute().print();

        //2.创建一个表用来保存 dwd_traffic_appvideo 的信息，并且加入处理时间，将现在这个表作为一个基表
        tEnv.executeSql("create table appvideo(" +
                "os string," +
                "ch string," +
                "is_new string," +
                "play_sec bigint," +
                "mid string," +
                "vc string," +
                "sid string," +
                "sc string," +
                "ar string," +
                "uid string," +
                "position_sec bigint," +
                "md string," +
                "video_id string," +
                "ba string," +
                "ts bigint," +
                "`pt` as proctime()" + //从这里拿实时的时间，给维度表进行lookup join提供处理时间 ，类似ods_db的操作
                ")" + SQLUtil.getKafkaSource(TOPIC_DWD_TRAFFIC_APPVIDEO,"Dwd_01_Play_VideoDetail")
        );
//        tEnv.sqlQuery("select * from appvideo").execute().print();

        //3.join
        Table result = tEnv.sqlQuery("select " +
                " av.video_id ," +
                " av.play_sec," +
                " av.uid," +
                " av.position_sec," +
                " vi.during_sec," +
                " cast(vi.chapter_id as string) chapter_id," +
                " av.ts," +
                " pt," +
                " current_row_timestamp()  as row_op_ts " +
                " from appvideo av " +
                " join video_info for system_time as of av.pt as vi on cast(av.video_id as bigint) = vi.id "
        );

        //4.创建表，并且连接到kafka内存
        tEnv.executeSql("create table dwd_play_video_detail(" +
                " video_id  string," +
                " play_sec bigint," +
                " uid string," +
                " position_sec bigint," +
                " during_sec bigint," +
                " chapter_id string," +
                " ts bigint," +
                " pt TIMESTAMP_LTZ(3)," +
                " row_op_ts timestamp_ltz(3) " +
                ")" + SQLUtil.getKafkaSink(TOPIC_DWD_PLAY_VIDEO_DETAIL)
        );

        result.executeInsert("dwd_play_video_detail");

    }
}
